package com.inns.position.hbase.service.impl;

import com.alibaba.fastjson.JSON;
import com.inns.position.hbase.HbaseConsts;
import com.inns.position.hbase.StatType;
import com.inns.position.hbase.bean.StatisticsResult;
import com.inns.position.hbase.config.properties.InnsProperties;
import com.inns.position.hbase.coprocessor.service.ObserverStatisticsProtos;
import com.inns.position.hbase.service.IHBaseService;
import com.inns.position.hbase.service.StatisticsService;
import com.inns.position.hbase.util.HbaseUtil;
import com.inns.position.hbase.util.StatisticsUtil;
import com.inns.position.kafka.bean.LocationRealTimeInfo;
import com.inns.position.redis.service.RealTimeDataService;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;

/**
 * Created by phoenix on 2018/4/6.
 */
@Service
public class StatisticsServiceRedisImpl implements StatisticsService {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Resource
    private IHBaseService ihBaseService;

    @Resource
    private RealTimeDataService realTimeDataService;

    private static final String familyColumn = HbaseConsts.FamilyColumn;

    private Connection connection;

    private ObserverStatisticsProtos.ExpandCell IMSICell = HbaseConsts.IMSICell;
    private ObserverStatisticsProtos.ExpandCell buildIdCell = HbaseConsts.buildIdCell;
    private ObserverStatisticsProtos.ExpandCell poiIdCell = HbaseConsts.poiIdCell;

    public void initTest(IHBaseService ihBaseService) throws IOException {
        this.ihBaseService = ihBaseService;
        init();
    }

    @PostConstruct
    public void init() throws IOException {
        connection = HbaseUtil.getConnection();
    }

    @Override
    public List<Cell> get(String table, String family, String rowKey, String qualify) {
        return ihBaseService.getRowKey(table, rowKey, family, qualify);
    }

    @Override
    public List<List<Cell>> scanRegexRowKey(String table, String regexKey) {
        return ihBaseService.scanRegexRowKey(table, regexKey);
    }

    @Override
    public void statistics331(StatType statType, long startTime, long endTime) {
        try {
            logger.info("statistics331:startTime=" + startTime + ",endTime=" + endTime);
            List<StatisticsResult> resultList = new ArrayList<>();

            Set<String> buildIds = realTimeDataService.getBuildIds();
            for (String buildId : buildIds) {
                int userCount = 0;
                long sumTime = 0;
                Set<String> imsiSet = new HashSet<>();
                Map<String, Long> buildImsiListByTime = realTimeDataService.getBuildImsiListByTime(buildId, startTime);
                for (String imsi : buildImsiListByTime.keySet()) {
                    LocationRealTimeInfo realTimeInfoByIMSI = realTimeDataService.getRealTimeInfoByIMSI(buildId, imsi);
                    long totalAccessTime = realTimeInfoByIMSI.getTotalAccessTime(startTime, endTime);
                    if (totalAccessTime > 0) {
                        userCount++;
                        sumTime += totalAccessTime;
                        imsiSet.add(imsi);
                    }
                }
                StatisticsResult result = new StatisticsResult();
                result.setGroupKey(buildId);
                result.setCount(userCount);
                result.setSum(sumTime);
                Map<String, Set<String>> distinctValues = new HashMap<>();
                distinctValues.put("IMSI", imsiSet);
                result.setDistinctValues(distinctValues);

                Map<String, Integer> countMap = new HashMap<>();
                countMap.put("IMSI", userCount);
                result.setCountMap(countMap);
                resultList.add(result);
            }

            StatisticsUtil.printResultList(resultList);

            saveStatResult(statType, startTime, endTime, resultList, HbaseConsts.TableName_Statistics_BuildId, false);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    @Override
    public void statistics332(StatType statType, long startTime, long endTime) {
        try {
            logger.info("statistics332:startTime=" + startTime + ",endTime=" + endTime);
            List<StatisticsResult> resultList = new ArrayList<>();

            Set<String> buildIds = realTimeDataService.getBuildIds();
            for (String buildId : buildIds) {
                Map<Integer, Long[]> poiMap = new HashMap<>();
                Set<String> imsiSet = new HashSet<>();

                Map<String, Long> buildImsiListByTime = realTimeDataService.getBuildImsiListByTime(buildId, startTime);
                for (String imsi : buildImsiListByTime.keySet()) {
                    Map<String, LocationRealTimeInfo> realTimeInfoPoiByIMSI = realTimeDataService.getRealTimeInfoPoiByIMSI(buildId, imsi);
                    for (LocationRealTimeInfo realTimeInfoByIMSI : realTimeInfoPoiByIMSI.values()) {
                        long totalAccessTime = realTimeInfoByIMSI.getTotalAccessTime(startTime, endTime);
                        if (totalAccessTime > 0) {
                            Long[] longs = poiMap.get(realTimeInfoByIMSI.getPoiId());
                            if (longs == null) {
                                longs = new Long[]{0L, 0L};
                            }
                            longs[0] = longs[0] + 1;
                            longs[1] = longs[1] + totalAccessTime;
                            imsiSet.add(imsi);
                            poiMap.put(realTimeInfoByIMSI.getPoiId(), longs);
                        }
                    }
                }
                for (Map.Entry<Integer, Long[]> s : poiMap.entrySet()) {
                    StatisticsResult result = new StatisticsResult();
                    result.setGroupKey(buildId + "|" + s.getKey());
                    result.setCount(s.getValue()[0].intValue());
                    result.setSum(s.getValue()[1]);
                    Map<String, Set<String>> distinctValues = new HashMap<>();
                    distinctValues.put("IMSI", imsiSet);
                    result.setDistinctValues(distinctValues);

                    Map<String, Integer> countMap = new HashMap<>();
                    countMap.put("IMSI", result.getCount());
                    result.setCountMap(countMap);
                    resultList.add(result);
                }
            }

            StatisticsUtil.printResultList(resultList);

            saveStatResult(statType, startTime, endTime, resultList, HbaseConsts.TableName_Statistics_POI, true);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    private void saveStatResult(StatType statType, long startTime, long endTime, List<StatisticsResult> resultList, String tableName, boolean poi) {
        for (StatisticsResult result : resultList) {
            String buildId = result.getSubKey(0);
            Set<String> imsiSet = result.getDistinctValues().get("IMSI");
            int newCount = newCount(buildId, endTime, imsiSet);

            Map<String, String> columnValues = new HashMap<>();
            columnValues.put("buildId", buildId);
            if (poi) {
                columnValues.put("poiId", result.getSubKey(1));
            }
            columnValues.put("newCount", newCount + "");
            columnValues.put("count", result.getCountMap().get("IMSI") + "");
            columnValues.put("totalTime", result.getSum() + "");
            columnValues.put("imsiSet", JSON.toJSONString(imsiSet));
            columnValues.put("startTime", startTime + "");
            columnValues.put("endTime", endTime + "");
            String rowKey = StatisticsUtil.getRowKey(statType, result.getGroupKey(), endTime);
            ihBaseService.putRowValueBatch(tableName, rowKey, familyColumn, endTime, columnValues);
        }
    }

    private int newCount(String buildId, long time, Set<String> imsiSet) {
        int count = 0;
        for (String imsi : imsiSet) {
            if (realTimeDataService.isNewUser(imsi, buildId, time)) {
                count++;
            }
        }
        return count;
    }

    @Override
    public void statistics333(StatType statType, long startTime, long endTime) {
        try {
            if (statType.getType() < StatType.DAY.getType()) {
                logger.error("statistics333 StatType must bigger then Day!");
                return;
            }
            logger.info("statistics333:startTime=" + startTime + ",endTime=" + endTime);
            List<StatisticsResult> resultList = new ArrayList<>();

            Set<String> buildIds = realTimeDataService.getBuildIds();
            for (String buildId : buildIds) {
                Map<String, Long[]> poiMap = new HashMap<>();

                Map<String, Long> buildImsiListByTime = realTimeDataService.getBuildImsiListByTime(buildId, startTime);
                for (String imsi : buildImsiListByTime.keySet()) {
                    Map<String, LocationRealTimeInfo> realTimeInfoPoiByIMSI = realTimeDataService.getRealTimeInfoPoiByIMSI(buildId, imsi);
                    for (LocationRealTimeInfo realTimeInfoByIMSI : realTimeInfoPoiByIMSI.values()) {
                        long totalAccessTime = realTimeInfoByIMSI.getTotalAccessTime(startTime, endTime);
                        if (totalAccessTime > 0) {
                            Long[] longs = poiMap.get(imsi + "|" + realTimeInfoByIMSI.getPoiId());
                            if (longs == null) {
                                longs = new Long[]{0L, 0L};
                            }
                            longs[0] = longs[0] + 1;
                            longs[1] = longs[1] + totalAccessTime;
                            poiMap.put(imsi + "|" + realTimeInfoByIMSI.getPoiId(), longs);
                        }
                    }
                }
                for (Map.Entry<String, Long[]> s : poiMap.entrySet()) {
                    StatisticsResult result = new StatisticsResult();
                    result.setGroupKey(buildId + "|" + s.getKey());
                    result.setCount(s.getValue()[0].intValue());
                    result.setSum(s.getValue()[1]);
                    resultList.add(result);
                }
            }

            StatisticsUtil.printResultList(resultList);
            for (StatisticsResult result : resultList) {
                Map<String, String> columnValues = new HashMap<>();
                columnValues.put("buildId", result.getSubKey(0));
                columnValues.put("IMSI", result.getSubKey(1));
                columnValues.put("poiId", result.getSubKey(2));
                columnValues.put("num", result.getCount() + "");
                columnValues.put("totalTime", result.getSum() + "");
                columnValues.put("startTime", startTime + "");
                columnValues.put("endTime", endTime + "");
                String rowKey = StatisticsUtil.getRowKey(statType, result.getGroupKey(), endTime);
                ihBaseService.putRowValueBatch(HbaseConsts.TableName_Statistics_User, rowKey, familyColumn, endTime, columnValues);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    @Override
    public void statistics334(StatType statType, long startTime, long endTime) {
        try {
            if (statType.getType() < StatType.DAY.getType()) {
                logger.error("statistics334 StatType must bigger then Day!");
                return;
            }
            logger.info("statistics334:startTime=" + startTime + ",endTime=" + endTime);
            List<StatisticsResult> resultList = new ArrayList<>();

            Set<String> buildIds = realTimeDataService.getBuildIds();
            for (String buildId : buildIds) {
                Map<String, Long[]> poiMap = new HashMap<>();

                Map<String, Long> buildImsiListByTime = realTimeDataService.getBuildImsiListByTime(buildId, startTime);
                for (String imsi : buildImsiListByTime.keySet()) {
                    LocationRealTimeInfo realTimeInfoByIMSI = realTimeDataService.getRealTimeInfoByIMSI(buildId, imsi);
                    long totalAccessTime = realTimeInfoByIMSI.getTotalAccessTime(startTime, endTime);
                    if (totalAccessTime > 0) {
                        Long[] longs = poiMap.get(imsi);
                        if (longs == null) {
                            longs = new Long[]{0L, 0L};
                        }
                        longs[0] = longs[0] + 1;
                        longs[1] = longs[1] + totalAccessTime;
                        poiMap.put(imsi, longs);
                    }
                }
                for (Map.Entry<String, Long[]> s : poiMap.entrySet()) {
                    StatisticsResult result = new StatisticsResult();
                    result.setGroupKey(buildId + "|" + s.getKey());
                    result.setCount(s.getValue()[0].intValue());
                    result.setSum(s.getValue()[1]);
                    resultList.add(result);
                }
            }
            // 统计
            List<Integer> result = getDurationStatistics(resultList);
            System.out.println("总记录数：" + resultList.size());
            System.out.println(result);
            {
                String buildId = resultList.get(0).getSubKey(0);

                Map<String, String> columnValues = new HashMap<>();
                columnValues.put("buildId", buildId);
                columnValues.put("intervals", HbaseConsts.intervalList.toString());
                columnValues.put("results", result.toString());
                columnValues.put("startTime", startTime + "");
                columnValues.put("endTime", endTime + "");
                String rowKey = StatisticsUtil.getRowKey(statType, buildId, endTime);
                ihBaseService.putRowValueBatch(HbaseConsts.TableName_Statistics_Duration_Frag_BuildId, rowKey, familyColumn, endTime, columnValues);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    private List<Integer> getDurationStatistics(List<StatisticsResult> resultList) {
        List<Float> intervalList = HbaseConsts.intervalList;
        List<Integer> result = new ArrayList<>();
        int[][] intervals = new int[intervalList.size() + 1][];
        int left = 0;
        for (int i = 0; i < intervalList.size(); i++) {
            int min = left;
            int max = (int) (intervalList.get(i) * 60 * 60 * 1000);//小时转毫秒
            left = max;
            intervals[i] = new int[]{min, max};
            result.add(0);
        }
        intervals[intervalList.size()] = new int[]{left, Integer.MAX_VALUE};
        result.add(0);
        // 输出
        for (StatisticsResult map : resultList) {
            long t = map.getSum();
            for (int i = 0; i < intervals.length; i++) {
                int[] a = intervals[i];
                if (a[0] <= t && t < a[1]) {
                    int s = result.remove(i);
                    result.add(i, s + 1);
                }
            }
        }
        return result;
    }

    @Override
    public void statistics334_poi(StatType statType, long startTime, long endTime) {
        try {
            if (statType.getType() < StatType.DAY.getType()) {
                logger.error("statistics334_poi StatType must bigger then Day!");
                return;
            }
            logger.info("statistics334_poi:startTime=" + startTime + ",endTime=" + endTime);
            List<StatisticsResult> resultList = new ArrayList<>();

            Set<String> buildIds = realTimeDataService.getBuildIds();
            for (String buildId : buildIds) {
                Map<String, Long[]> poiMap = new HashMap<>();

                Map<String, Long> buildImsiListByTime = realTimeDataService.getBuildImsiListByTime(buildId, startTime);
                for (String imsi : buildImsiListByTime.keySet()) {
                    Map<String, LocationRealTimeInfo> realTimeInfoPoiByIMSI = realTimeDataService.getRealTimeInfoPoiByIMSI(buildId, imsi);
                    for (LocationRealTimeInfo realTimeInfoByIMSI : realTimeInfoPoiByIMSI.values()) {
                        long totalAccessTime = realTimeInfoByIMSI.getTotalAccessTime(startTime, endTime);
                        if (totalAccessTime > 0) {
                            Long[] longs = poiMap.get(imsi + "|" + realTimeInfoByIMSI.getPoiId());
                            if (longs == null) {
                                longs = new Long[]{0L, 0L};
                            }
                            longs[0] = longs[0] + 1;
                            longs[1] = longs[1] + totalAccessTime;
                            poiMap.put(imsi + "|" + realTimeInfoByIMSI.getPoiId(), longs);
                        }
                    }
                }
                for (Map.Entry<String, Long[]> s : poiMap.entrySet()) {
                    StatisticsResult result = new StatisticsResult();
                    result.setGroupKey(buildId + "|" + s.getKey());
                    result.setCount(s.getValue()[0].intValue());
                    result.setSum(s.getValue()[1]);
                    resultList.add(result);
                }
            }

            // 统计
            List<Integer> result = getDurationStatistics(resultList);
            System.out.println("总记录数：" + resultList.size());
            System.out.println(result);
            {
                String buildId = resultList.get(0).getSubKey(0);
                String poiId = resultList.get(0).getSubKey(1);

                Map<String, String> columnValues = new HashMap<>();
                columnValues.put("buildId", buildId);
                columnValues.put("poiId", poiId);
                columnValues.put("intervals", HbaseConsts.intervalList.toString());
                columnValues.put("results", result.toString());
                columnValues.put("startTime", startTime + "");
                columnValues.put("endTime", endTime + "");
                String rowKey = StatisticsUtil.getRowKey(statType, buildId + "|" + poiId, endTime);
                ihBaseService.putRowValueBatch(HbaseConsts.TableName_Statistics_Duration_Frag_POI, rowKey, familyColumn, endTime, columnValues);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    @Override
    public void statistics335(StatType statType, long startTime, long endTime) {
        try {
            if (statType.getType() < StatType.WEEK.getType()) {
                logger.error("statistics335 StatType must bigger then WEEK!");
                return;
            }
            logger.info("statistics335:startTime=" + startTime + ",endTime=" + endTime);
            List<StatisticsResult> resultList = new ArrayList<>();

            Set<String> buildIds = realTimeDataService.getBuildIds();
            for (String buildId : buildIds) {
                Map<String, Long> buildImsiListByTime = realTimeDataService.getBuildImsiListByTime(buildId, startTime);
                for (String imsi : buildImsiListByTime.keySet()) {
                    LocationRealTimeInfo realTimeInfoByIMSI = realTimeDataService.getRealTimeInfoByIMSI(buildId, imsi);
                    int count = realTimeInfoByIMSI.getAccessTimesByTimeInterval(startTime, endTime).size();
                    if (count > 0) {
                        StatisticsResult result = new StatisticsResult();
                        result.setGroupKey(buildId + "|" + imsi);
                        result.setCount(count);
                        resultList.add(result);
                    }
                }
            }

            // 统计
            List<Integer> result = getFrequencyStatistics(resultList);
            System.out.println("总记录数：" + resultList.size());
            System.out.println(result);
            {
                String buildId = resultList.get(0).getSubKey(0);

                Map<String, String> columnValues = new HashMap<>();
                columnValues.put("buildId", buildId);
                columnValues.put("results", result.toString());
                columnValues.put("startTime", startTime + "");
                columnValues.put("endTime", endTime + "");
                String rowKey = StatisticsUtil.getRowKey(statType, buildId, endTime);
                ihBaseService.putRowValueBatch(HbaseConsts.TableName_Statistics_Frequency_Frag_BuildId, rowKey, familyColumn, endTime, columnValues);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    @Override
    public void statistics335_poi(StatType statType, long startTime, long endTime) {
        try {
            if (statType.getType() < StatType.WEEK.getType()) {
                logger.error("statistics335_poi StatType must bigger then WEEK!");
                return;
            }
            logger.info("statistics335_poi:startTime=" + startTime + ",endTime=" + endTime);
            List<StatisticsResult> resultList = new ArrayList<>();

            Set<String> buildIds = realTimeDataService.getBuildIds();
            for (String buildId : buildIds) {
                Map<Integer, Integer> poiMap = new HashMap<>();
                Map<String, Long> buildImsiListByTime = realTimeDataService.getBuildImsiListByTime(buildId, startTime);
                for (String imsi : buildImsiListByTime.keySet()) {
                    Map<String, LocationRealTimeInfo> realTimeInfoPoiByIMSI = realTimeDataService.getRealTimeInfoPoiByIMSI(buildId, imsi);
                    for (LocationRealTimeInfo realTimeInfoByIMSI : realTimeInfoPoiByIMSI.values()) {
                        int count = realTimeInfoByIMSI.getAccessTimesByTimeInterval(startTime, endTime).size();
                        if (count > 0) {
                            Integer integer = poiMap.get(realTimeInfoByIMSI.getPoiId());
                            if (integer == null) {
                                poiMap.put(realTimeInfoByIMSI.getPoiId(), 1);
                            } else {
                                poiMap.put(realTimeInfoByIMSI.getPoiId(), integer + 1);
                            }
                        }
                    }
                }
                for (Map.Entry<Integer, Integer> e : poiMap.entrySet()) {
                    StatisticsResult result = new StatisticsResult();
                    result.setGroupKey(buildId + "|" + e.getKey());
                    result.setCount(e.getValue());
                    resultList.add(result);
                }
            }

            // 统计
            List<Integer> result = getFrequencyStatistics(resultList);
            System.out.println("总记录数：" + resultList.size());
            System.out.println(result);
            {
                String buildId = resultList.get(0).getSubKey(0);
                String poiId = resultList.get(0).getSubKey(1);

                Map<String, String> columnValues = new HashMap<>();
                columnValues.put("buildId", buildId);
                columnValues.put("poiId", poiId);
                columnValues.put("results", result.toString());
                columnValues.put("startTime", startTime + "");
                columnValues.put("endTime", endTime + "");
                String rowKey = StatisticsUtil.getRowKey(statType, buildId + "|" + poiId, endTime);
                ihBaseService.putRowValueBatch(HbaseConsts.TableName_Statistics_Frequency_Frag_POI, rowKey, familyColumn, endTime, columnValues);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    private List<Integer> getFrequencyStatistics(List<StatisticsResult> resultList) {
        int maxTimes = InnsProperties.instance().getMaxTimes();
        List<Integer> result = new ArrayList<>(maxTimes + 1);
        for (int i = 0; i <= maxTimes; i++) {
            result.add(0);
        }

        for (StatisticsResult map : resultList) {
            int count = map.getCount();
            int index = count - 1;
            if (count > maxTimes) {
                index = maxTimes;
            }
            int s = result.remove(index);
            result.add(index, s + 1);
        }
        return result;
    }

    @Override
    public void statistics336(StatType statType, long startTime, long endTime) {
        try {
            logger.info("statistics336:startTime=" + startTime + ",endTime=" + endTime);
//            List<StatisticsResult> resultList = StatisticsUtil.getGroupAndCount(getResults(startTime, endTime), Arrays.asList(IMSICell, buildIdCell/*, poiIdCell*/), Arrays.asList(buildIdCell));
            List<StatisticsResult> resultList = new ArrayList<>();
            StatisticsUtil.printResultList(resultList);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    @Override
    public void statPath(StatType statType, long startTime, long endTime) {
        ResultScanner scanner = null;
        try {
            logger.info("statPath:startTime=" + startTime + ",endTime=" + endTime);
            long l = System.currentTimeMillis();
            Table table = connection.getTable(TableName.valueOf(HbaseConsts.TableName_Original));
            Scan scan = new Scan();
            scan.setTimeRange(startTime, endTime);
            scanner = table.getScanner(scan);

            Map<String, List<String>> resultMap = new HashMap<>();
            for (Result result : scanner) {
                List<Cell> rowCells = result.listCells();
                dealRowCells(rowCells, resultMap);
            }
            logger.info("总记录数：" + resultMap.size() + ",useTime=" + (System.currentTimeMillis() - l) / 1000);
            System.out.println(resultMap);

            // 存库
            for (Map.Entry<String, List<String>> e : resultMap.entrySet()) {
                String imsi = StatisticsResult.getSubKey(e.getKey(), 0);
                String buildId = StatisticsResult.getSubKey(e.getKey(), 1);

                Map<String, String> columnValues = new HashMap<>();
                columnValues.put("buildId", buildId);
                columnValues.put("IMSI", imsi);
                columnValues.put("results", JSON.toJSONString(e.getValue()));
                String rowKey = StatisticsUtil.getRowKey(statType, buildId + "|" + imsi, endTime);
                ihBaseService.putRowValueBatch(HbaseConsts.TableName_Statistics_Path, rowKey, familyColumn, endTime, columnValues);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (scanner != null) {
                    scanner.close();
                }
            } catch (Exception ignored) {
            }
        }
    }

    private void dealRowCells(List<Cell> rowCells, Map<String, List<String>> resultMap) {
        List<ObserverStatisticsProtos.ExpandCell> groupColumnsList = Arrays.asList(IMSICell, buildIdCell);

        Map<String, String> groupKeys = new HashMap<>();
        String x = "", y = "";
        long timestemp = 0;
        continueTag:
        for (Cell cell : rowCells) {
            timestemp = cell.getTimestamp();
            String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
            String qualify = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
            for (ObserverStatisticsProtos.ExpandCell gCell : groupColumnsList) {
                if (family.equals(gCell.getFamily().toStringUtf8()) && qualify.equals(gCell.getQualify().toStringUtf8())) {
                    String groupValue = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    groupKeys.put(gCell.getQualify().toStringUtf8(), groupValue);
                    continue continueTag;
                }
            }
            if (family.equals("info") && qualify.equals("x")) {
                x = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
            } else if (family.equals("info") && qualify.equals("y")) {
                y = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
            }
        }
        String groupKey = "";
        for (ObserverStatisticsProtos.ExpandCell gCell : groupColumnsList) {
            groupKey = groupKey + groupKeys.get(gCell.getQualify().toStringUtf8()) + "|";
        }

        List<String> paths = resultMap.get(groupKey);
        if (paths == null) {
            paths = new ArrayList<>();
            resultMap.put(groupKey, paths);
        }
        paths.add(timestemp + "," + x + "," + y);
    }

}
